package com.algorithm.example.queue.blockingQueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <p>
 * 双锁实现
 * </p>
 *
 * @author lzt
 * @version 1.0
 * @since 2024/1/28 20:11
 */
public class BlockingQueue2<E> implements BlockingQueue<E> {

    private final E[] array;
    private int head;
    private int tail;

    // 原子类
    private AtomicInteger size;
    private ReentrantLock headLock = new ReentrantLock();
    private ReentrantLock tailLock = new ReentrantLock();
    private Condition headWaits = headLock.newCondition();
    private Condition tailWaits = tailLock.newCondition();

    public BlockingQueue2(int capacity) {
        this.array = (E[]) new Object[capacity];
    }

    @Override
    public void offer(E e) throws InterruptedException {
        int c; // 添加前元素个数
        // 加锁
        tailLock.lockInterruptibly();
        try {
            while (isFull()) {
                // 等待唤醒
                tailWaits.await();
            }
            array[tail] = e;
            if (++tail >= array.length) {
                tail = 0;
            }
            //获取当前的值并自增
            c = size.getAndIncrement();

            if (c + 1 < array.length) {
                tailWaits.signal();
            }
        } finally {
            // 解锁
            tailLock.unlock();
        }

        // 防止死锁
        if (c == 0) {
            headLock.lockInterruptibly();
            try {
                // 唤醒线程
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }

    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        // 加锁
        tailLock.lockInterruptibly();
        // 毫秒 -> 纳米
        long t = TimeUnit.MILLISECONDS.toNanos(timeout);
        try {
            while (isFull()) {
                if (t <= 0) {
                    return false;
                }
                // 等待唤醒
                t = tailWaits.awaitNanos(t);
            }
            array[tail] = e;
            if (++tail > array.length) {
                tail = 0;
            }
            size.incrementAndGet();

            if (size.get() + 1 < array.length) {
                tailWaits.signal();
            }

        } finally {
            // 解锁
            tailLock.unlock();
        }
        headLock.lockInterruptibly();
        try {
            // 唤醒线程
            headWaits.signal();
        } finally {
            headLock.unlock();
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        int c;
        headLock.lockInterruptibly();
        E e;
        try {
            while (isEmpty()) {
                headWaits.await();
            }
            e = array[head];
            array[head] = null;
            if (++head >= array.length) {
                head = 0;
            }
            c = size.getAndDecrement();
            if (c > 1) {
                headWaits.signal();
            }
        } finally {
            headLock.unlock();
        }
        if (c == array.length) {
            tailLock.lockInterruptibly();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }

        return e;
    }

    /**
     * 队列容量是否已满
     *
     * @return 满返回true 否则返回false
     */
    public boolean isFull() {
        return size.get() == array.length;
    }

    /**
     * 队列是否为空
     *
     * @return 空返回true 否则返回false
     */
    public boolean isEmpty() {
        return size.get() == 0;
    }

    @Override
    public String toString() {
        return "BlockingQueue1{" +
                "array=" + Arrays.toString(array) +
                '}';
    }
}
